package main

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"time"
)

// 发送消息
func Producter() {
	p, err := rocketmq.NewProducer(
		producer.WithNameServer([]string{"127.0.0.1:9876"}),
		producer.WithGroupName("mxshop"))
	if err != nil {
		panic("生成producer失败 ")
	}
	if err = p.Start(); err != nil {
		panic("启动producer失败 ")
	}
	res, err := p.SendSync(context.Background(), primitive.NewMessage("2104A", []byte("色狼")))
	if err != nil {
		fmt.Printf("发送失败 : %s\n", err)
	} else {
		fmt.Printf("发送成功 : %s\n", res.String())
	}
	if err = p.Shutdown(); err != nil {
		panic("关闭producer失败 ")
	}
}

// 消费消息
func main() {
	m, _ := rocketmq.NewPushConsumer(consumer.WithNameServer([]string{"127.0.0.1:9876"}),
		consumer.WithGroupName("mxshop"))
	err := m.Subscribe("2104A", consumer.MessageSelector{}, func(ctx context.Context, ext ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for i := range ext {
			fmt.Printf("获取到值：  %v \n", ext[i])
		}
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {
		panic(err)
	}
	_ = m.Start()
	time.Sleep(time.Hour)
	_ = m.Shutdown()
}

// 延时消息
func LaterMessage() {
	p, err := rocketmq.NewProducer(producer.WithNameServer([]string{"192.168.124.51:9876"}))
	if err != nil {
		panic("生成producer失败 ")
	}

	if err = p.Start(); err != nil {
		panic("启动producer失败 ")
	}

	msg := primitive.NewMessage("2104A", []byte("给你"))
	msg.WithDelayTimeLevel(3)
	res, err := p.SendSync(context.Background(), msg)
	if err != nil {
		fmt.Printf("发送失败 : %s\n", err)
	} else {
		fmt.Printf("发送成功 : %s\n", res.String())
	}

	if err = p.Shutdown(); err != nil {
		panic("关闭producer失败 ")
	}

	//支付的时候，  淘宝，  12306，  购票，  超时归还  - 定时执行逻辑
	//我可以去写一个轮询，  轮询的问题：  1. 多久执行一次轮询  30分钟
	//在12:00执行过一次，  下一次执行就是在  12:30的时候  但是12:01的时候下了单，  12:31就应该超时
	//那我1分钟执行一次啊，  比如我的订单量没有这么大， 1分钟执行一次，  其中29次查询都是无用，  而且你还还会轮询mysql
	//rocketmq的延迟消息，  1. 时间一到就执行，  2. 消息中包含了订单编号，你只查询这种订单编号
}

type OrderListener struct{}

func (o *OrderListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
	//成功执行
	fmt.Println("开始执行本地逻辑 ")
	time.Sleep(time.Second * 3)
	fmt.Println("执行本地逻辑成功 ")
	return primitive.CommitMessageState

	//执行失败回滚
	//fmt.Println("开始执行本地逻辑 ")
	//time.Sleep(time.Second * 3)
	//fmt.Println("执行本地逻辑失败回滚 ")
	//return primitive.Rollback MessageState

	//执行失败
	//fmt.Println("开始执行本地逻辑 ")
	//time.Sleep(time.Second * 3)
	//fmt.Println("执行本地逻辑失败 ")
	//本地执行逻辑无缘无故失败  代码异常  宕机
	//return primitive.UnknowState
}

func (o *OrderListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
	fmt.Println("rocketmq的消息回查 ")
	time.Sleep(time.Second * 15)
	return primitive.CommitMessageState
}

// 事务消息
func TrancationMessage() {
	p, err := rocketmq.NewTransactionProducer(
		&OrderListener{},
		producer.WithNameServer([]string{"192.168.124.51:9876"}),
	)
	if err != nil {
		panic("生成producer失败 ")
	}
	if err = p.Start(); err != nil {
		panic("启动producer失败 ")
	}
	res, err := p.SendMessageInTransaction(context.Background(), primitive.NewMessage("2014A", []byte("你好")))
	if err != nil {
		fmt.Printf("发送失败 : %s\n", err)
	} else {
		fmt.Printf("发送成功 : %s\n", res.String())
	}
	time.Sleep(time.Hour)
	if err = p.Shutdown(); err != nil {
		panic("关闭producer失败 ")
	}
}
